package org.tao.lightningmq.broker.config;

import com.alibaba.fastjson.serializer.SerializerFeature;
import com.alibaba.fastjson2.JSON;
import org.tao.lightningmq.broker.cache.CommonCache;
import org.tao.lightningmq.broker.model.ConsumeQueueOffsetModel;
import org.tao.lightningmq.broker.utils.FileContentUtil;
import org.tao.lightningmq.common.constants.BrokerConstants;

import java.util.concurrent.TimeUnit;

/**
 * @Author lht
 * @date 2025/1/4 - 22:43
 * @description: 消费队列配置加载器
 */
public class ConsumeQueueOffsetLoader {

    String filePath;

    public void loadProperties() {
        GlobalProperties globalProperties = CommonCache.getGlobalProperties();
        String mqHome = globalProperties.getMqHome();
        filePath = mqHome + "/config/consume-queue-offset.json";
        String jsonContent = FileContentUtil.readFromFile(filePath);
        ConsumeQueueOffsetModel offsetModel = JSON.parseObject(jsonContent, ConsumeQueueOffsetModel.class);
        CommonCache.setConsumeQueueOffsetModel(offsetModel);
    }

    public void startRefreshConsumeQueueOffsetTask() {
        CommonThreadPoolConfig.setRefreshConsumeQueueOffsetTask(() -> {
            while (true) {
                try {
                    TimeUnit.SECONDS.sleep(BrokerConstants.DEFAULT_REFRESH_CONSUME_QUEUE_OFFSET_TIME_STEP);
                    ConsumeQueueOffsetModel consumeQueueOffsetModel = CommonCache.getConsumeQueueOffsetModel();
                    FileContentUtil.writeToFile(filePath, com.alibaba.fastjson.JSON.toJSONString(consumeQueueOffsetModel, SerializerFeature.PrettyFormat));
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });
    }

}
